package com.example.biyesheji.controller;

import com.example.biyesheji.mapper.ProductMapper;
import com.example.biyesheji.model.NewRecodeParam;
import com.example.biyesheji.pojo.ProductSales;
import com.example.biyesheji.service.NewRecodeService;
import com.example.biyesheji.utils.MapUtil;
import lombok.SneakyThrows;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 商品控制器
 *
 * @author xwh
 * @date 2023/02/08
 */
@RestController
@CrossOrigin
@RequestMapping("/product")
public class commodityController {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private JavaSparkContext sc;

    @Autowired
    private NewRecodeService newRecodeService;

/*
    @Autowired
    private ScheduledExecutorService scheduledExecutorService;

    //加载commodity
    List<String> commodityList = new ArrayList<>();
    //计数器
    Long i = 0l;
    //sse
    SseEmitter sseEmitter =new SseEmitter();

    @GetMapping("/commodity")
    public SseEmitter getCommodity(){

        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {
                i++;
                //调用查询方法
                List<String> category = productMapper.getCategory(i);
                //赋值计算
                commodityList.add(category.get(0));
                //RDD加载内存
                JavaRDD<String> rdd = sc.parallelize(commodityList);
                //RDD格式转换
                JavaPairRDD<String, Integer> javaPairRDD = rdd.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
                //计算
                JavaPairRDD<String, Integer> mapJavaPairRDD = javaPairRDD.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
                //行动操作
                Map<String, Integer> map = mapJavaPairRDD.collectAsMap();

                Map<String, Integer> map1 = MapUtil.sortDescend(map);

                sseEmitter.send(map1.entrySet().stream().findFirst().get().getKey());
            }
        },10,5, TimeUnit.SECONDS);

        return sseEmitter;
    }
*/
    @GetMapping("/commodity")
    public String getCommodity(){
        //查数据
        List<String> list = productMapper.getcategoryCodeList();
        //System.out.println(list.size());
        //RDD加载内存
        JavaRDD<String> rdd = sc.parallelize(list);
        //RDD格式转换
        JavaPairRDD<String, Integer> javaPairRDD = rdd.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        //计算
        JavaPairRDD<String, Integer> mapJavaPairRDD = javaPairRDD.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        //行动操作
        Map<String, Integer> map = mapJavaPairRDD.collectAsMap();

        Map<String, Integer> map1 = MapUtil.sortDescend(map);
        map1.remove(null);

        return map1.entrySet().iterator().next().getKey();
    }


    @GetMapping("/age")
    public Integer getAge(){
        List<Integer> ageList = productMapper.getAgeList();
        //RDD计算总值
        //内存中加载数据
        JavaRDD<Integer> rdd = sc.parallelize(ageList);

        //RDD格式转换
        JavaPairRDD<Integer, Integer> javaPairRDD = rdd.mapToPair((PairFunction<Integer, Integer, Integer>) integer -> new Tuple2<>(integer,1));
        //计算
        JavaPairRDD<Integer, Integer> mapJavaPairRDD = javaPairRDD.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        //行动操作
        Map<Integer, Integer> map = mapJavaPairRDD.collectAsMap();

        Map<Integer, Integer> map1 = MapUtil.sortDescend(map);
        map1.remove(null);
        rdd.unpersist();
        return map1.entrySet().iterator().next().getKey();
    }

    /**
     * @Params
     * @Return @return {@link Long }
     * @Description getquantity
     * @Throws
     * @Author xwh
     * @Date 2023/02/10
     */
    @GetMapping("/quantity")
    public Long getquantity(){
        return productMapper.orderQuenty();
    }


    /**
     * @Params
     * @Return @return {@link List }<{@link NewRecodeParam }>
     * @Description 获得最近数据
     * @Throws
     * @Author xwh
     * @Date 2023/02/11
     */
    @GetMapping("/newRecord")
    public List<NewRecodeParam> getNewRecode(){
        return newRecodeService.getNewRecode();
    }

}
